package com.demo.spark;

import java.util.Arrays;
import java.util.Iterator;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.SparkSession;

import scala.Tuple2;

public class TestSpark {

	@SuppressWarnings("serial")
	public static void main(String[] args) {

		SparkConf conf = new SparkConf();
		conf.setAppName("Spark by java.");
		conf.setMaster("spark://192.168.6.131:7077");

		JavaSparkContext sc = new JavaSparkContext(conf);

		JavaRDD<String> textFile = sc.textFile("hdfs://192.168.6.131:9000/spark/spark.txt");

		JavaRDD<String> words = textFile.flatMap(new FlatMapFunction<String, String>() {

			@Override
			public Iterator<String> call(String soruce) throws Exception {
				return Arrays.asList(soruce.split(" ")).iterator();
			}
		});

		JavaPairRDD<String, Integer> wordPair = words.mapToPair(new PairFunction<String, String, Integer>() {

			@Override
			public Tuple2<String, Integer> call(String soruce) throws Exception {
				return new Tuple2<String, Integer>(soruce, 1);
			}
		});

		JavaPairRDD<String, Integer> wordCount = wordPair.reduceByKey(new Function2<Integer, Integer, Integer>() {

			@Override
			public Integer call(Integer v1, Integer v2) throws Exception {
				return v1 + v2;
			}
		});
		
		// JavaPairRDD<String, Integer> counts = textFile
		// .flatMap(s -> Arrays.asList(s.split(" ")).iterator())
		// .mapToPair(word -> new Tuple2<>(word, 1))
		// .reduceByKey((a, b) -> a + b);
		
		wordCount.saveAsTextFile("hdfs://192.168.6.131:9000/spark/aaa.txt");
		sc.close();
	}

}
